package lab.poc.erl;

import java.util.Iterator;
import java.util.List;
import java.util.Map.Entry;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.SynchronousQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.locks.ReentrantLock;

public final class Node {

	private static final Node vmNode = new Node("localVM");

	private final String nodeName;
	private /**/final int creation = 0; // revisit: increment on each Node restart
	private int serial = 0;
	private int pidCount = 1;
	private final int[] refIds = {1, 0, 0, 0};

	private final ConcurrentHashMap<String, Pid> registry = new ConcurrentHashMap<String, Pid>();
	private final ReentrantLock nodeLock = new ReentrantLock(true); // fair ordering policy

	// Threadpool for Processes/Actors
	final ExecutorService actorService = createActorService();

	public static Node getNode() {
		return vmNode;
	}

	private Node(final String nodeName) {
		this.nodeName = nodeName;
	}

	public void register(final String alias, final Pid pid) {
		if (registry.putIfAbsent(alias, pid) != null) {
			throw new IllegalArgumentException(alias + " is already registered");
		}
	}

	public void unregister(final String alias) {
		registry.remove(alias);
	}

	public Object send(final String alias, final Object msg) {
		Pid pid = whereIs(alias);
		if (Pid.UNDEFINED != pid) {
			return pid.send(msg);
		}
		throw new IllegalArgumentException(alias + " is not registered");
	}

	public Pid whereIs(final String alias) {
		Pid pid = registry.get(alias);
		if (pid != null) {
			return pid;
		}
		return Pid.UNDEFINED;
	}

	final void deregister(final Pid pid) {
		for (Iterator<Entry<String, Pid>> it = registry.entrySet().iterator(); it.hasNext(); ) {
			Entry<String, Pid> e = it.next();
			if (e.getValue().equals(pid)) {
				it.remove();
			}
		}
	}

	public Pid spawn(final Class<? extends Actor> actorClass) {
		return spawn(actorClass, null);
	}

	public Pid spawnLink(final Class<? extends Actor> actorClass) {
		return spawnLink(actorClass, null);
	}

	public Pid spawn(final Class<? extends Actor> actorClass,
			final List<Object> args) {
		return createPid(actorClass, args, false);
	}

	public Pid spawnLink(final Class<? extends Actor> actorClass,
			final List<Object> args) {
		return createPid(actorClass, args, true);
	}

	public String getName() {
		return nodeName;
	}

	private Pid createPid(final Class<? extends Actor> actorClass,
			final List<Object> args, final boolean linkToCreator) {
		if (actorClass == null) {
			throw new IllegalArgumentException("null actorClass");
		}		
		nodeLock.lock();
		try {
			final Pid pid = new Pid(this, actorClass, args, pidCount, serial,
					creation, linkToCreator);
			++pidCount;
			if (pidCount > 0x7fff) {
				pidCount = 0;
				++serial;
				if (serial > 0x1fff) { /* 13 bits */
					serial = 0;
				}
			}
			return pid;
		} finally {
			nodeLock.unlock();
		}
	}

	public final Reference makeRef() {
		synchronized (refIds) {
			final Reference ref = new Reference(nodeName, refIds, creation);

			++refIds[0];
			if (refIds[0] < 0) {
				refIds[0] = 0;

				++refIds[1];
				if (refIds[1] < 0) {
					refIds[1] = 0;

					++refIds[2];
					if (refIds[2] < 0) {
						refIds[2] = 0;

						++refIds[3];
						if (refIds[3] < 0) {
							refIds[3] = 0;
						}
					}
				}
			}
			return ref;
		}
	}

	public final Pid createMockProcessForCurrentThread() {
		Pid self = Pid.PID.get();
		if (self == null) {
			return createPid(MockProcess.class, null, false);
		}
		if (!self.isMockPid()) {
			throw new IllegalThreadStateException(
					"Actor must not create a mock process (it is already a process)");
		}
		return self;
	}

	public final void removeMockProcessForCurrentThread() {
		Pid self = Pid.PID.get();
		if (self != null && self.isMockPid()) {
			Pid.PID.remove();
			Actor.MBOX.remove();
		}
	}

	private static ExecutorService createActorService() {
		// comparable to Executors.newCachedThreadPool(threadFactory), except
		// for TPExecutor, CallerRunsPolicy and corePoolSize = 60
		final ExecutorService workerService = new TPExecutor(
				60 /* NR_START_THREADS */,
				Integer.MAX_VALUE /* NR_MAX_THREADS */,
				60000L /* KEEP_ALIVE_TIME */, TimeUnit.MILLISECONDS,
				new SynchronousQueue<Runnable>(), new ThreadFactory() {
					private final AtomicLong counter = new AtomicLong();

					@Override
					public Thread newThread(final Runnable r) {
						Thread t = new Thread(r, "ActorThread-"
								+ counter.getAndIncrement());
						t.setDaemon(true);
						return t;
					}
				}, new ThreadPoolExecutor.CallerRunsPolicy());
		return workerService;
	}
}
